package com.zhxy.util.publish;

import java.io.Serializable;

import com.zhxy.util.async.ResultHandler;
import com.zhxy.util.message.AsyncMessageProcessor;
import com.zhxy.util.message.DuplexMessagePipe;
import com.zhxy.util.message.DuplexMessageService;
import com.zhxy.util.message.DuplexMessageServiceListener;
import com.zhxy.util.message.MessagePipe;

public class DuplexMessagePublisherRouter<P> extends AbstractPublisher<P> implements Publisher<P> {
	
	private DuplexMessageService service;
	
	private volatile MessagePipe clientPipe;
	
	private SimplePublisherRouter<P> router = new SimplePublisherRouter<P>();
	
	private DuplexMessageServiceListener connectionListener = new DuplexMessageServiceListener() {
		public void disconnected(DuplexMessagePipe pipe) {
			// 连接断开后应该清理该连接所有的订阅信息
			@SuppressWarnings("unchecked")
			PipeSubscriber ps = (PipeSubscriber)pipe.getAttribute(PipeSubscriber.class.getName());
			router.cleanSubscribe(ps);
		}
		public void connected(DuplexMessagePipe pipe) {
			clientPipe = pipe;
			// 为每个连接只创建唯一一个PipeSubscriber
			pipe.setAttribute(PipeSubscriber.class.getName(), new PipeSubscriber(pipe));
		}
	}; 
	
	public void setPublisher(Publisher<P> publisher) {
		router.setPublisher(publisher);
	}
	
	public void attach(DuplexMessageService service, boolean serverMode) {
		if (this.service != null)
			throw new IllegalStateException("已经attach到service");
		this.service = service;
		if (serverMode) {
			this.service.addMessageProcessor(new ServerProcessor());
		} else {
			this.service.addMessageProcessor(new ClientProcessor());
		}
		this.service.addDuplexMessageServiceListener(connectionListener);
	}
	
	public void detach() {
		if (this.service == null)
			throw new IllegalStateException("没有attach到service");
		
		service.removeDuplexMessageServiceListener(connectionListener);
	}
	
	@Override
	public void subscribe(Subscriber<P> subscriber, String name)
			throws SubscribeException {
		doSubscribe(subscriber, name);
		SubscribeMessage msg = new SubscribeMessage();
		msg.name = name;
		clientPipe.send(msg);
	}

	@Override
	public void unsubscribe(Subscriber<P> subscriber, String name)
			throws SubscribeException {
		doUnsubscribe(subscriber, name);
	}

	@Override
	public void cleanSubscribe(Subscriber<P> subscriber)
			throws SubscribeException {
		doCleanSubscribe(subscriber);
	}
	
	public static class Message implements Serializable {
		private static final long serialVersionUID = -642071042809146108L;
	}
	public static class SubscribeMessage extends Message {
		private static final long serialVersionUID = -2340659499177304805L;
		private String name;
	}
	public static class UnsubscribeMessage extends Message {
		private static final long serialVersionUID = -6760735102304607488L;
		private String name;
	}
	public static class PublishMessage<P> extends Message {
		private static final long serialVersionUID = -7228735134593350239L;
		private String name;
		private P pub;
	}
	
	/**
	 * 为每个连接只创建一个PipeSubscriber
	 * @author 刘勉
	 *
	 */
	class PipeSubscriber implements Subscriber<P> {
		private MessagePipe pipe;
		public PipeSubscriber(MessagePipe pipe) {
			this.pipe = pipe;
		}
		@Override
		public void feed(String name, P pub) {
			PublishMessage<P> msg = new PublishMessage<P>();
			msg.name = name;
			msg.pub = pub;
			pipe.send(msg);
		}
	}
	
	
	class ServerProcessor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {
		@SuppressWarnings("unchecked")
		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			if (message instanceof SubscribeMessage) {
				SubscribeMessage msg = (SubscribeMessage)message;
				PipeSubscriber ps = (PipeSubscriber)context.getAttribute(PipeSubscriber.class.getName());
				router.subscribe(ps, msg.name);
			} else if (message instanceof UnsubscribeMessage) {
				UnsubscribeMessage msg = (UnsubscribeMessage)message;
				PipeSubscriber ps = (PipeSubscriber)context.getAttribute(PipeSubscriber.class.getName());
				router.unsubscribe(ps, msg.name);
			}
			
			if (handler != null)
				handler.result(new Message());
		}
	}
	
	class ClientProcessor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {
		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			if (message instanceof PublishMessage) {
				@SuppressWarnings("unchecked")
				PublishMessage<P> msg = (PublishMessage<P>)message;
				doPublish(msg.name, msg.pub);
			}
			
			if (handler != null)
				handler.result(new Message());
			
		}
	}
}
